package cn.dfun.sample.flink.tabletest

import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.DataTypes
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.descriptors._
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.scala._

// es upsert模式
// 重点测聚合
// curl "node-01:9200/_cat/indices"
object EsOutputTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val tableEnv = StreamTableEnvironment.create(env)

    val inputPath = "C:\\wor\\flink-sample\\src\\main\\resources\\sensor"
    tableEnv.connect(new FileSystem().path(inputPath))
      // 旧版,非标,弃用不支持kafka
      //        .withFormat(new OldCsv())
      .withFormat(new Csv())
      .withSchema(new Schema()
        .field("id", DataTypes.STRING())
        .field("timestamp", DataTypes.BIGINT())
        .field("temp", DataTypes.DOUBLE())
      )
      .createTemporaryTable("inputTable")
    // 3 转换操作
    val sensorTable = tableEnv.from("inputTable")
    // 3.1 简单转换
    val resultTable = sensorTable
      .select('id, 'temp)
      .filter('id === "sensor_1")
    // 3.2 聚合转换
    val aggTable = sensorTable
      .groupBy('id) // 基于id分组
      .select('id, 'id.count as 'count)

    // 输出到es
    tableEnv.connect(new Elasticsearch()
        .version("6")
        .host("node-01", 9200, "http")
        .index("sensor2")
        .documentType("temperature")
        )
        // upsert模式
        .inUpsertMode()
        .withFormat(new Json())
        .withSchema(new Schema()
            .field("id", DataTypes.STRING())
            .field("count", DataTypes.BIGINT())
        )
        .createTemporaryTable("esOutputTable")
    aggTable.insertInto("esOutputTable")
    env.execute("es output test")
  }
}
